package ooo.reindeer.concurrent.netqueue.handler;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import ooo.reindeer.concurrent.netqueue.IMessageHandler;
import ooo.reindeer.concurrent.netqueue.message.Command;
import ooo.reindeer.concurrent.netqueue.message.Message;
import ooo.reindeer.logging.ILogger;
import ooo.reindeer.logging.LogUtil;

import java.util.concurrent.ConcurrentHashMap;

@ChannelHandler.Sharable
public class QueueServerMessageHandler extends SimpleChannelInboundHandler<Message> {

    private static final ILogger logger = LogUtil.getLogger(QueueServerMessageHandler.class);
    private final ConcurrentHashMap<String, IMessageHandler> queueMap = new ConcurrentHashMap<String, IMessageHandler>();

    public IMessageHandler cancelRegiste(String topic) {
        return queueMap.remove(topic);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        for (String s : queueMap.keySet()) {

            Message registe = new Message();
            registe.setCommand(Command.REGISTE);
            registe.setTopic(s);
            ctx.writeAndFlush(registe).sync();
        }
        logger.debug("queue server active [{}]", ctx);
//        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        logger.debug("queue server Inactive [{}]", ctx);
//        ctx.fireChannelInactive();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Message message) throws Exception {

        if (message.getCommand() == Command.PUSH) {
            logger.debug("PUSH {}", message);
            IMessageHandler handler = queueMap.get(message.getTopic());
            Message answer = new Message();
            answer.setCommand(Command.PUSH_ANSWER);
            answer.setTopic(message.getTopic());
            if (handler == null) {
                answer.setStatus("NOHDL");
            } else {
                if (handler.pushHandle(message)) {
                    answer.setStatus("SUCCE");
                } else {
                    answer.setStatus("ERROR");
                }
            }
            channelHandlerContext.writeAndFlush(answer);

        } else if (message.getCommand() == Command.REGISTE_ANSWER) {
            logger.debug("REGISTE_ANSWER {}", message);
        } else if (message.getCommand() == Command.CANCEL_REGISTE_ANSWER) {
            logger.debug("CANCEL_REGISTE_ANSWER {}", message);
            queueMap.remove(message.getTopic());
        } else if (message.getCommand() == Command.PUT_ANSWER) {
            logger.debug("PUT_ANSWER {}", message);
        } else {
            logger.debug("UNKNOW {}", message);
        }

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error("exception [{}] :{}", ctx, cause);
        ctx.close();
    }

    public void registe(String topic, IMessageHandler handler) {
        queueMap.put(topic, handler);
    }
}
